package cn.com.lyb.flink.transfrom;

import cn.com.lyb.flink.bean.WaterSensor;
import cn.com.lyb.flink.functions.MyMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 算子  map 映射
 * 我们只需要基于DataStream调用map()方法就可以进行转换处理。方法需要传入的参数是接口MapFunction的实现；返回值类型还是DataStream，不过泛型（流中的元素类型）可能改变。
 */
public class MapDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<WaterSensor> sensorDS = env.fromElements(new WaterSensor("s1", 1L, 1),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s3", 3L, 3));
        // 方式1 匿名实现类
//        sensorDS.map(new MapFunction<WaterSensor, String>() {
//            @Override
//            public String map(WaterSensor waterSensor) throws Exception {
//                return waterSensor.getId();
//            }
//        }).print();

        // 方式2 lambda
        // sensorDS.map((MapFunction<WaterSensor, String>) waterSensor -> waterSensor.getId()).print();

        // 方式3 定义一个类实现MapFunction
        sensorDS.map(new MyMapFunction()).print();

        env.execute();
    }
}
